-module(erl_sql_compare_db).
-feature(maybe_expr, enable).
-author("zhuiyi9009@163.com").

-include("common.hrl").
-include_lib("emysql/include/emysql.hrl").

%% API
-export([init_db/2, execute/2, fetch/2]).
-export([pool_id/1]).

init_db(PoolId, Dsn) ->
    maybe
        {ok, {User, Password, Host, Port0, Database, Params}} ?= dsn_parser:parse_dsn(Dsn),
        {ok, _} ?= application:ensure_all_started(emysql),

        Encoding = list_to_atom(proplists:get_value("encoding", Params, "utf8")),
        Port = list_to_integer(Port0),

        ok ?=
        case emysql_conn_mgr:has_pool(PoolId) of
            true ->
                ok;
            _ ->
                ?DEBUG("connect mysql, User ~p, Password ~p, Host ~p, Port ~p, Database ~p, Encoding ~p~n", [User, Password, Host, Port, Database, Encoding]),
                emysql:add_pool(PoolId, 1, User, Password, Host, Port, undefined, Encoding)
        end,

        {ok, _} ?= ensure_database(PoolId, Database),
        ok
    else
        {error, Reason} ->
            ?WARN("init_db error ~p~n", [Reason]),
            {error, Reason};
        Other ->
            ?WARN("init_db unknown error ~p~n", [Other]),
            {error, Other}
    end.

ensure_database(PoolId, Database) ->
    Sql = io_lib:format("SHOW DATABASES LIKE '~s';", [Database]),
    case fetch(PoolId, Sql) of
        {ok, [_]} ->
            execute(PoolId, io_lib:format("USE ~s;", [Database]));
        {ok, []} ->
            ?WARN("~s not exists, create database~n", [Database]),
            case execute(PoolId, io_lib:format("CREATE DATABASE ~s;", [Database])) of
                {ok, _} ->
                    execute(PoolId, io_lib:format("USE ~s;", [Database]));
                Error ->
                    Error
            end;
        Error ->
            Error
    end.

execute(PoolId, Sql) ->
    case emysql:execute(PoolId, Sql) of
        #error_packet{} = Result ->
            print_error(Sql, Result);
        #result_packet{} ->
            ok;
        #ok_packet{affected_rows = AffectedRows} ->
            {ok, AffectedRows};
        [_|_] ->
            %% 多个sql的结果
            ok
    end.

fetch(PoolId, Sql) ->
    case emysql:execute(PoolId, Sql) of
        #result_packet{rows = [undefined]} ->
            %% todo 这里的Row可能会返回[undefined], 不知道是从哪个版本开始
            {ok, []};
        #result_packet{rows = Rows} ->
            {ok, Rows};
        Result when is_record(Result, error_packet) ->
            print_error(Sql, Result)
    end.

print_error(Sql, #error_packet{msg = Msg}) ->
    ?WARN("Sql: [~s], error: ~s~n", [Sql, Msg]),
    {db_error, Msg}.

pool_id(#schema_sync{side = ?COMPARE_SIDE_SOURCE}) ->
    ?EMYSQL_SOURCE_POOL_ID;
pool_id(#schema_sync{side = ?COMPARE_SIDE_DEST}) ->
    ?EMYSQL_DEST_POOL_ID.